package com.xiaoze.transfrom;

import com.xiaoze.bean.WaterSensor;
import com.xiaoze.function.MyFilterFunctionImpl;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author 小泽
 * @create 2024-09-02  15:16
 * filter 过滤算子
 * 记得每天敲代码哦
 */
public class FilterDemo {
       public static void main(String[] args) throws Exception {
              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
              // 设置并行度为1
              env.setParallelism(1);
              // 构建元数据
              DataStreamSource<WaterSensor> sensorDs = env.fromElements(
                      new WaterSensor("s1", 1L, 1),
                      new WaterSensor("s1", 111L, 1111),
                      new WaterSensor("s2", 2L, 2),
                      new WaterSensor("s3", 3L, 3)
              );

              // filter 算子，一条一条过滤
              // 方式一：匿名内部类
              // SingleOutputStreamOperator<WaterSensor> filter = sensorDs.filter(new FilterFunction<WaterSensor>() {
              //        @Override
              //        public boolean filter(WaterSensor value) throws Exception {
              //               return "s1".equals(value.getId());
              //        }
              // });

              // 方式二：lambda表达式
              // SingleOutputStreamOperator<WaterSensor> filter = sensorDs.filter(sensor -> "s1".equals(sensor.getId()));

              // 方式三：自定义一个类来实现filterFunction接口
              SingleOutputStreamOperator<WaterSensor> filter = sensorDs.filter(new MyFilterFunctionImpl());

              filter.print();

              env.execute();
       }
}
